package com.stars.easyms.mq.core;

import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.service.MQMessageService;
import com.stars.easyms.redis.lock.DistributedLock;
import com.stars.easyms.base.util.DateTimeUtil;
import lombok.extern.slf4j.Slf4j;

import java.sql.Timestamp;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MQ消息调度者线程(发送、接收各一个单例线程)：
 * 1.生成发送/接收处理分类者线程的线程池
 * 2.抓取发送/接收处理失败的数据
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
@Slf4j
public final class MQMigrator implements Runnable {

    private static MQMigrator instance;

    private static final int PER_MAX_MOVE_SIZE = 1500;

    private final DistributedLock schedulerClearLock = new DistributedLock(MQConstants.MQ_SCHEDULER_LOCK, "C", 60);

    private final AtomicInteger loopExecuteExceptionCount = new AtomicInteger();

    private final MQMessageService mqMessageService;

    @Override
    public void run() {

        // 循环执行mq调度任务,当失败次数过多时停止执行
        while (loopExecuteExceptionCount.get() < MQConstants.MAX_SCHEDULE_EXCEPTION_COUNT) {
            try {
                Thread.sleep(clearOldData() ? 43200000 + ThreadLocalRandom.current().nextLong(43200000) : 10000);
            } catch (Exception e) {
                log.error("MQMigrator execute failure!", e);
                loopExecuteExceptionCount.incrementAndGet();
            }
        }
    }

    private boolean clearOldData() {
        boolean lockFlag = false;
        try {
            // 1.为了防止分布式服务器同时读取同一份数据，因此需要加锁
            lockFlag = schedulerClearLock.lock();
            if (!lockFlag) {
                return false;
            }

            // 2.清理7天以前的旧数据
            Map<String, Object> paramMap = new HashMap<>(4);
            paramMap.put("status", MQConstants.MQ_STATUS_SUCCESS);
            paramMap.put("size", PER_MAX_MOVE_SIZE);
            paramMap.put("fireTime", new Timestamp(DateTimeUtil.getDateByOneDayOffset(new Date(), -7).getTime()));
            int insertCount;
            do {
                insertCount = mqMessageService.insertMQMessageOldData(paramMap);
                if (insertCount > 0) {
                    paramMap.put("size", insertCount);
                    mqMessageService.deleteMQMessageOldData(paramMap);
                }
            } while (insertCount == PER_MAX_MOVE_SIZE);
            return true;
        } finally {
            if (lockFlag) {
                schedulerClearLock.unlock();
            }
        }
    }

    public static MQMigrator createInstance(MQMessageService mqMessageService) {
        MQMigrator localInstance = instance;
        if (localInstance == null) {
            synchronized (MQMigrator.class) {
                localInstance = instance;
                if (localInstance == null) {
                    instance = localInstance = new MQMigrator(mqMessageService);
                }
            }
        }
        return localInstance;
    }

    private MQMigrator(MQMessageService mqMessageService) {
        this.mqMessageService = mqMessageService;
    }

}